package io.kiki.sba.registry.server.clientInterface.push;


import io.kiki.sba.registry.common.model.DataCenterPushInfo;
import io.kiki.sba.registry.common.model.SegmentPushInfo;
import io.kiki.sba.registry.common.model.SubscriberUtils;
import io.kiki.sba.registry.common.model.client.pb.MultiReceivedDataPb;
import io.kiki.sba.registry.common.model.client.pb.MultiSegmentDataPb;
import io.kiki.sba.registry.common.model.client.pb.ReceivedDataPb;
import io.kiki.sba.registry.common.model.store.*;
import io.kiki.sba.registry.core.model.Data;
import io.kiki.sba.registry.core.model.MultiReceivedData;
import io.kiki.sba.registry.core.model.ReceivedConfigData;
import io.kiki.sba.registry.core.model.ReceivedData;
import io.kiki.sba.registry.exception.SofaRegistryRuntimeException;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.converter.ReceivedDataConverter;
import io.kiki.sba.registry.server.clientInterface.converter.pb.ReceivedDataConvertor;
import io.kiki.sba.registry.server.clientInterface.converter.pb.ReceivedDataConvertor.CompressorGetter;
import io.kiki.sba.registry.server.clientInterface.multi.cluster.DataCenterCache;
import io.kiki.sba.registry.server.clientInterface.predicate.ZonePredicate;
import io.kiki.sba.registry.server.clientInterface.providedata.CompressPushService;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.StringUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.Map.Entry;
import java.util.function.Predicate;

public class PushDataGenerator {

    @Autowired
    Config config;

    @Resource
    CompressPushService compressPushService;

    @Autowired
    DataCenterCache dataCenterCache;

    public PushData createPushData(DataCenterToSubDatumMap dataCenterToSubDatumMap, Map<String, Subscriber> subscriberMap) {
        dataCenterToSubDatumMap.mustUnzipped();
        if (subscriberMap.size() > 1) {
            SubscriberUtils.getAndAssertHasSameScope(subscriberMap.values());
            SubscriberUtils.getAndAssertAcceptedEncodes(subscriberMap.values());
            SubscriberUtils.getAndAssertAcceptMulti(subscriberMap.values());
        }
        // only supported 4.x
        SubscriberUtils.assertClientVersion(subscriberMap.values(), BaseInfo.ClientVersion.StoreData);

        final Subscriber subscriber = subscriberMap.values().iterator().next();
        String clientCell = config.getClientCell(subscriber.getZone());

        CompressorGetter compressorGetter = (Map<String, List<Data>> data) -> compressPushService.getCompressor(data, subscriber.getAcceptEncodes(), subscriber.getClientUrl().getIp());
        if (subscriber.isAcceptMulti()) {
            return createMultiReceivePushData(dataCenterToSubDatumMap, clientCell, new ArrayList<>(subscriberMap.keySet()), subscriber, compressorGetter);
        }
        return createReceivePushData(dataCenterToSubDatumMap, clientCell, new ArrayList<>(subscriberMap.keySet()), subscriber, compressorGetter);
    }

    private PushData createMultiReceivePushData(DataCenterToSubDatumMap unzipDatum, String clientCell, List<String> subscriberRegisterIdList, Subscriber subscriber, CompressorGetter compressorGetter) {
        Predicate<String> pushDataPredicate = ZonePredicate.pushDataPredicate(unzipDatum.getDataId(), clientCell, subscriber.getScope(), config);

        Set<String> pushDataCenters = unzipDatum.getDataCenterToSubDatumMap().keySet();
        Map<String, Set<String>> segmentZones = dataCenterCache.getZoneSet(pushDataCenters);

        Set<String> metadataDataCenters = segmentZones.keySet();

        if (!pushDataCenters.equals(metadataDataCenters)) {
            throw new SofaRegistryRuntimeException(StringFormatter.format("createMultiReceivePushData error, datum.dataCenters: {}, metadata.dataCenters: {}", pushDataCenters, metadataDataCenters));
        }
        PushData<MultiReceivedData> pushData = ReceivedDataConverter.getMultiReceivedData(unzipDatum, subscriber.getScope(), subscriberRegisterIdList, clientCell, config.getDataCenter(), pushDataPredicate, segmentZones);

        final Byte serializerIndex = subscriber.getClientUrl().getSerializerIndex();
        if (serializerIndex == null || Url.PROTOBUF != serializerIndex) {
            return pushData;
        }

        StringUtil.checkNotEmpty(pushData.getPayload().getMultiData(), "multiSegmentDatas");
        MultiReceivedDataPb multiReceivedDataPb = ReceivedDataConvertor.convert2MultiPb(pushData.getPayload(), compressorGetter);

        fillSegmentPushInfo(pushData, multiReceivedDataPb);

        return new PushData<>(multiReceivedDataPb, pushData.getDataCenterPushInfo());
    }

    private void fillSegmentPushInfo(PushData<MultiReceivedData> pushData, MultiReceivedDataPb multiReceivedDataPb) {
        for (Entry<String, MultiSegmentDataPb> entry : multiReceivedDataPb.getMultiDataMap().entrySet()) {
            pushData.addSegmentInfo(entry.getKey(), entry.getValue().getEncoding(), entry.getValue().getZipData().size());
        }
    }

    private PushData createReceivePushData(DataCenterToSubDatumMap unzipDatum, String clientCell, List<String> subscriberRegisterIdList, Subscriber subscriber, CompressorGetter compressorGetter) {
        Predicate<String> pushDataPredicate = ZonePredicate.pushDataPredicate(unzipDatum.getDataId(), clientCell, subscriber.getScope(), config);

        PushData<ReceivedData> pushData = ReceivedDataConverter.getReceivedData(unzipDatum, subscriber.getScope(), subscriberRegisterIdList, clientCell, config.getDataCenter(), pushDataPredicate);

        final Byte serializerIndex = subscriber.getClientUrl().getSerializerIndex();
        if (serializerIndex == null || Url.PROTOBUF != serializerIndex) {
            return pushData;
        }

        StringUtil.nullValue(pushData.getPayload().getData(), "datas");
        ReceivedDataPb receivedDataPb = ReceivedDataConvertor.convert2Pb(pushData.getPayload(), compressorGetter);

        if (receivedDataPb.getBody() == null || StringUtils.isEmpty(receivedDataPb.getEncoding())) {
            return new PushData(receivedDataPb, pushData.getDataCenterPushInfo());
        } else {
            pushData.addSegmentInfo(receivedDataPb.getSegment(), receivedDataPb.getEncoding(), receivedDataPb.getBody().size());
            return new PushData<>(receivedDataPb, pushData.getDataCenterPushInfo());
        }
    }

    public PushData createPushData(Watcher watcher, ReceivedConfigData data) {
        Url url = watcher.getClientUrl();
        Object o = data;
        if (url.getSerializerIndex() != null && Url.PROTOBUF == url.getSerializerIndex()) {
            o = ReceivedDataConvertor.convert2Pb(data);
        }

        String dataCenter = config.getDataCenter();
        DataCenterPushInfo dataCenterPushInfo = new DataCenterPushInfo();
        dataCenterPushInfo.setSegmentPushInfos(Collections.singletonMap(dataCenter, new SegmentPushInfo(dataCenter, 1)));
        return new PushData(o, Collections.singletonMap(dataCenter, dataCenterPushInfo));
    }
}
